package com.example.inventoryservice.service;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.pattern.Patterns;
import akka.util.Timeout;
import com.example.inventoryservice.actor.PersistentInventoryActor;
import com.example.inventoryservice.event.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CompletionStage;

/**
 * 订单事件消费者
 * 消费RocketMQ中的订单事件，并转换为内部命令发送给InventoryActor
 */
@Slf4j
@Service
public class OrderEventConsumer {

    @Autowired
    private ActorSystem actorSystem;

    @Autowired
    private ObjectMapper objectMapper;

    private final Timeout timeout = Timeout.create(Duration.ofSeconds(5));

    /**
     * 消费订单创建事件
     */
    @RocketMQMessageListener(
        topic = "ORDER_EVENTS",
        selectorExpression = "ORDER_CREATED",
        consumerGroup = "inventory-consumer-group"
    )
    public class OrderCreatedEventConsumer implements RocketMQListener<String> {

        @Override
        public void onMessage(String message) {
            try {
                log.info("收到订单创建事件: {}", message);
                
                // 解析事件
                OrderCreatedEvent event = objectMapper.readValue(message, OrderCreatedEvent.class);
                
                // 创建PersistentInventoryActor
                ActorRef inventoryActor = actorSystem.actorOf(PersistentInventoryActor.props(), "inventory-" + event.getOrderId());
                
                // 构建预留库存命令
                PersistentInventoryActor.ReserveInventory reserveCmd = new PersistentInventoryActor.ReserveInventory(
                    event.getOrderId(),
                    event.getItems()
                );
                
                // 发送命令并处理响应
                CompletionStage<Object> future = Patterns.ask(inventoryActor, reserveCmd, timeout);
                
                future.thenAccept(response -> {
                    PersistentInventoryActor.InventoryReserved result = (PersistentInventoryActor.InventoryReserved) response;
                    if ("SUCCESS".equals(result.getStatus())) {
                        log.info("库存预留成功: 订单ID={}, 预留详情={}", 
                            result.getOrderId(), result.getReservedItems());
                    } else {
                        log.warn("库存预留失败: 订单ID=" + result.getOrderId() + ", 状态=" + result.getStatus());
                    }
                }).exceptionally(throwable -> {
                    log.error("处理订单创建事件异常: {}", throwable.getMessage(), throwable);
                    return null;
                });
                
            } catch (Exception e) {
                log.error("消费订单创建事件失败: {}", e.getMessage(), e);
            }
        }
    }

    /**
     * 消费订单支付事件
     */
    @RocketMQMessageListener(
        topic = "ORDER_EVENTS",
        selectorExpression = "ORDER_PAID",
        consumerGroup = "inventory-consumer-group"
    )
    public class OrderPaidEventConsumer implements RocketMQListener<String> {

        @Override
        public void onMessage(String message) {
            try {
                log.info("收到订单支付事件: {}", message);
                
                // 解析事件
                OrderPaidEvent event = objectMapper.readValue(message, OrderPaidEvent.class);
                
                // 创建PersistentInventoryActor
                ActorRef inventoryActor = actorSystem.actorOf(PersistentInventoryActor.props(), "inventory-" + event.getOrderId());
                
                // 构建确认库存命令
                PersistentInventoryActor.ConfirmInventory confirmCmd = new PersistentInventoryActor.ConfirmInventory(
                    event.getOrderId()
                );
                
                // 发送命令并处理响应
                CompletionStage<Object> future = Patterns.ask(inventoryActor, confirmCmd, timeout);
                
                future.thenAccept(response -> {
                    PersistentInventoryActor.InventoryConfirmed result = (PersistentInventoryActor.InventoryConfirmed) response;
                    if ("SUCCESS".equals(result.getStatus())) {
                        log.info("库存确认成功: 订单ID={}", result.getOrderId());
                    } else {
                        log.warn("库存确认失败: 订单ID=" + result.getOrderId() + ", 状态=" + result.getStatus());
                    }
                }).exceptionally(throwable -> {
                    log.error("处理订单支付事件异常: {}", throwable.getMessage(), throwable);
                    return null;
                });
                
            } catch (Exception e) {
                log.error("消费订单支付事件失败: {}", e.getMessage(), e);
            }
        }
    }

    /**
     * 消费订单取消事件
     */
    @RocketMQMessageListener(
        topic = "ORDER_EVENTS",
        selectorExpression = "ORDER_CANCELLED",
        consumerGroup = "inventory-consumer-group"
    )
    public class OrderCancelledEventConsumer implements RocketMQListener<String> {

        @Override
        public void onMessage(String message) {
            try {
                log.info("收到订单取消事件: {}", message);
                
                // 解析事件
                OrderCancelledEvent event = objectMapper.readValue(message, OrderCancelledEvent.class);
                
                // 创建PersistentInventoryActor
                ActorRef inventoryActor = actorSystem.actorOf(PersistentInventoryActor.props(), "inventory-" + event.getOrderId());
                
                // 构建释放库存命令
                PersistentInventoryActor.ReleaseInventory releaseCmd = new PersistentInventoryActor.ReleaseInventory(
                    event.getOrderId()
                );
                
                // 发送命令并处理响应
                CompletionStage<Object> future = Patterns.ask(inventoryActor, releaseCmd, timeout);
                
                future.thenAccept(response -> {
                    PersistentInventoryActor.InventoryReleased result = (PersistentInventoryActor.InventoryReleased) response;
                    if ("SUCCESS".equals(result.getStatus())) {
                        log.info("库存释放成功: 订单ID={}", result.getOrderId());
                    } else {
                        log.warn("库存释放失败: 订单ID=" + result.getOrderId() + ", 状态=" + result.getStatus());
                    }
                }).exceptionally(throwable -> {
                    log.error("处理订单取消事件异常: {}", throwable.getMessage(), throwable);
                    return null;
                });
                
            } catch (Exception e) {
                log.error("消费订单取消事件失败: {}", e.getMessage(), e);
            }
        }
    }
}
